package service

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/golang/protobuf/proto"
	"google.golang.org/grpc/status"
	"net/http"
	"prod/api/prod"
	"prod/internal/auth"
	"prod/internal/conf"
	"prod/internal/dao"
	"prod/pkg/ecode"
	"prod/pkg/log"
	"prod/pkg/network"
	"prod/pkg/nsq"
)

var (
	Server       = &Service{}
	processorMap = map[string]func(context.Context, []byte) error{
		//constant.TopicProdUpdate: Server.BatchUpdateProdsHandler,
		//constant.TopicProdDelete: Server.BatchDeleteProdsHandler,
	}
)

type Service struct {
	dao          *dao.Dao
	nsqConsumers []*nsq.Consumer
	server       *network.NetWork
	// 标记当前服务状态
	status bool
}

func (s *Service) Init(ctx context.Context) (err error) {
	// 初始化数据库...外部执行器
	if s.dao, err = dao.New(ctx, conf.Conf.ORM, conf.Conf.Cache, conf.Conf.Env, conf.Conf.Downstream); err != nil {
		return
	}
	// 初始化消息生产者
	if s.dao.NSQProducer, err = nsq.InitProducer(ctx, conf.Conf.NSQConfig); err != nil {
		return
	}
	// 初始化消息消费者
	if !conf.Conf.SkipConsume {
		if s.nsqConsumers, err = nsq.InitConsumers(
			ctx, conf.Conf.NSQConfig, conf.Conf.Env, conf.Conf.ServiceName, processorMap); err != nil {
			return
		}
	} else {
		s.nsqConsumers = make([]*nsq.Consumer, 0)
		log.FromContext(ctx).Infow("NSQ消费处理...跳过")
	}
	if s.server, err = network.New(ctx, conf.Conf.NetWorkConfig, conf.Conf.Env, conf.Conf.ServiceName,
		errorFormatter,
		conf.Conf.Auth.TokenInterceptor,
		network.CustomJsonOutput(),
		network.CustomHeaderMatcher(map[string]string{
			auth.JWTTokenHeaderKey: auth.JWTTokenMetaKey,
		}),
		network.CustomProtoErrorHandler(errorDefaultProcessor, errorFormatter)); err != nil {
		return
	}
	// 枚举自定义校验器
	if err = network.Validator.RegisterValidation(
		"in_enum", network.EnumValidator(conf.Conf.ServiceName)); err != nil {
		err = fmt.Errorf("枚举自定义校验器注册失败, error=%s", err)
	}
	prod.RegisterProdServer(s.server.RpcServer, s)
	if err = prod.RegisterProdHandlerClient(
		ctx, s.server.ServeMux, prod.NewProdClient(s.server.RpcVirtualClient)); err != nil {
		err = fmt.Errorf("RPC注册http执行器失败, error=%s", err)
		return
	}
	s.status = true
	return
}

// will block until signal
func (s *Service) StartServer() (err error) {
	if err = s.server.StartServer(); err != nil {
		return
	}
	return
}

func (s *Service) Close(ctx context.Context) {
	// 有序关闭服务
	if s.status {
		// 关闭http服务，不再接受请求
		log.FromContext(ctx).Infow("prod服务关闭...开始")
		if s.server != nil {
			s.server.ShutDown(ctx)
		}
		// 关闭消息执行器
		if len(s.nsqConsumers) != 0 {
			for _, c := range s.nsqConsumers {
				c.Stop()
			}
			log.FromContext(ctx).Info("NSQ所有Consumer已关闭")
		}
		// 关闭外部调用执行器
		if s.dao != nil {
			s.dao.Close(ctx)
		}
		log.FromContext(ctx).Infow("prod服务关闭...完成")
		s.status = false
	}
}

// 仅用于测试
func (s *Service) SetDao(d *dao.Dao) {
	s.dao = d
}

// 标准化http错误格式
func errorFormatter(e *ecode.ErrorCode) proto.Message {
	return &prod.DefaultResp{
		Status:       false,
		ErrorCode:    e.CodeMessage,
		ErrorMessage: e.ErrMessage,
		ErrorType:    e.ErrorType.ErrType,
	}
}

func errorDefaultProcessor(s *status.Status, w http.ResponseWriter) (writeFinished bool) {
	for _, detail := range s.Details() {
		if t, ok := detail.(*prod.DefaultResp); ok {
			if e := json.NewEncoder(w).Encode(t); e != nil {
				_ = json.NewEncoder(w).Encode(errorFormatter(ecode.ErrSystemError))
			}
			return true
		}
	}
	return false
}
